package simpledb.storage;

import simpledb.common.Database;
import simpledb.common.Debug;
import simpledb.transaction.TransactionId;

import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Constructor;
import java.util.*;

/**
 * 这个类实现了 SimpleDb 的恢复子系统，负责记录不同类型的日志记录，并确保遵循写前日志（Write-Ahead Logging, WAL）和两阶段锁定协议。
 * LogFile 类负责管理数据库的日志文件，它能够根据需要写入不同的日志记录类型,如事务的开始、提交、中止、更新等。
 * 这些日志记录对于数据库的崩溃恢复至关重要。调用者必须确保遵循写前日志和两阶段锁定协议。
 */
public class LogFile {

    final File               logFile;                              // 日志文件
    private RandomAccessFile raf;                                  // 随机访问文件对象
    Boolean                  recoveryUndecided   = true;           // 是否已决定进行恢复

    static final int         ABORT_RECORD        = 1;              // 中止记录类型
    static final int         COMMIT_RECORD       = 2;              // 提交记录类型
    static final int         UPDATE_RECORD       = 3;              // 更新记录类型
    static final int         BEGIN_RECORD        = 4;              // 开始记录类型
    static final int         CHECKPOINT_RECORD   = 5;              // 检查点记录类型
    static final long        NO_CHECKPOINT_ID    = -1;             // 表示没有检查点

    final static int         INT_SIZE            = 4;              // 整数大小
    final static int         LONG_SIZE           = 8;              // 长整数大小

    long                     currentOffset       = -1;             // 当前文件偏移量
    int                      totalRecords        = 0;              // 总记录数

    final Map<Long, Long>    tidToFirstLogRecord = new HashMap<>(); // 记录每个事务的第一个日志记录位置

    /**
     * 构造函数，初始化并支持指定文件的日志文件。
     *
     * @param f 日志文件名称
     */
    public LogFile(File f) throws IOException {
        this.logFile = f;
        raf = new RandomAccessFile(f, "rw");
        recoveryUndecided = true;

        // 安装关闭钩子以强制在关闭时清理
        // Runtime.getRuntime().addShutdownHook(new Thread() { public void run() { shutdown(); } });

        // 警告：这里没有任何东西可以验证指定的日志文件是否对应当前目录。
        // 这可能会导致问题，因为我们记录表ID，而这些ID可能与当前目录中的表ID不匹配。
    }

    /**
     * 在追加日志记录之前执行的操作。
     * 如果不确定数据库是否要进行恢复，则现在确定——它不会。因此截断日志。
     */
    void preAppend() throws IOException {
        totalRecords++;
        if (recoveryUndecided) {
            recoveryUndecided = false;
            raf.seek(0);
            raf.setLength(0);
            raf.writeLong(NO_CHECKPOINT_ID); // 写入没有检查点的标识
            raf.seek(raf.length());
            currentOffset = raf.getFilePointer();
        }
    }

    /**
     * 获取总记录数。
     *
     * @return 总记录数
     */
    public synchronized int getTotalRecords() {
        return totalRecords;
    }

    /**
     * 为指定的事务 ID 写入一个中止记录到日志中，强制日志刷新到磁盘，并执行回滚。
     *
     * @param tid 中止的事务
     */
    public void logAbort(TransactionId tid) throws IOException {
        synchronized (Database.getBufferPool()) {
            synchronized (this) {
                preAppend();
                rollback(tid);

                raf.writeInt(ABORT_RECORD);
                raf.writeLong(tid.getId());
                raf.writeLong(currentOffset);
                currentOffset = raf.getFilePointer();
                force();
                tidToFirstLogRecord.remove(tid.getId());
            }
        }
    }

    /**
     * 为指定的事务 ID 写入一个提交记录到日志中，并强制日志刷新到磁盘。
     *
     * @param tid 提交的事务
     */
    public synchronized void logCommit(TransactionId tid) throws IOException {
        preAppend();
        Debug.log("COMMIT " + tid.getId());

        raf.writeInt(COMMIT_RECORD);
        raf.writeLong(tid.getId());
        raf.writeLong(currentOffset);
        currentOffset = raf.getFilePointer();
        force();
        tidToFirstLogRecord.remove(tid.getId());
    }

    /**
     * 为指定的事务 ID 和页面写入一个更新记录到日志中（带有提供的前后图像）。
     *
     * @param tid  执行写操作的事务
     * @param before 页面的前图像
     * @param after 页面的后图像
     */
    public synchronized void logWrite(TransactionId tid, Page before, Page after) throws IOException {
        Debug.log("WRITE, offset = " + raf.getFilePointer());
        preAppend();

        raf.writeInt(UPDATE_RECORD);
        raf.writeLong(tid.getId());

        writePageData(raf, before); // 写入前图像数据
        writePageData(raf, after); // 写入后图像数据
        raf.writeLong(currentOffset);
        currentOffset = raf.getFilePointer();

        Debug.log("WRITE OFFSET = " + currentOffset);
    }

    /**
     * 将一个页面的数据写入到日志文件中
     * 首先获取页面ID并序列化，然后依次写入页面类名、ID类名、ID信息以及页面数据本身。
     * @param raf               随机访问文件对象
     * @param p                 要写入的页面对象
     * @throws IOException
     */
    void writePageData(RandomAccessFile raf, Page p) throws IOException {
        PageId pid = p.getId();
        int[] pageInfo = pid.serialize();

        // 页面数据格式：
        // 页面类名
        // ID类名
        // ID类字节
        // ID类数据
        // 页面类字节
        // 页面类数据

        String pageClassName = p.getClass().getName();
        String idClassName = pid.getClass().getName();

        raf.writeUTF(pageClassName);
        raf.writeUTF(idClassName);

        raf.writeInt(pageInfo.length);
        for (int j : pageInfo) {
            raf.writeInt(j);
        }
        byte[] pageData = p.getPageData();
        raf.writeInt(pageData.length);
        raf.write(pageData);
    }

    /**
     * 从日志文件中读取页面数据，并返回一个新的页面实例。
     * 读取类名、构造函数参数，利用反射创建相应的页面和ID对象。
     * @param raf
     * @return
     * @throws IOException
     */
    Page readPageData(RandomAccessFile raf) throws IOException {
        String pageClassName = raf.readUTF();
        String idClassName = raf.readUTF();

        try {
            Class<?> idClass = Class.forName(idClassName);
            Class<?> pageClass = Class.forName(pageClassName);

            Constructor<?>[] idConsts = idClass.getDeclaredConstructors();
            int numIdArgs = raf.readInt();
            Object[] idArgs = new Object[numIdArgs];
            for (int i = 0; i < numIdArgs; i++) {
                idArgs[i] = raf.readInt();
            }
            PageId pid = (PageId) idConsts[0].newInstance(idArgs);

            Constructor<?>[] pageConsts = pageClass.getDeclaredConstructors();
            int pageSize = raf.readInt();
            byte[] pageData = new byte[pageSize];
            raf.read(pageData);

            Object[] pageArgs = new Object[] { pid, pageData };
            return (Page) pageConsts[0].newInstance(pageArgs);
        } catch (Exception e) {
            e.printStackTrace();
            throw new IOException();
        }
    }

    /**
     * 为指定事务记录开始日志。
     * 检查是否已存在该事务的日志记录，不存在则写入BEGIN记录，并更新事务首次日志记录位置。
     * @param tid
     * @throws IOException
     */
    public synchronized void logXactionBegin(TransactionId tid) throws IOException {
        if (tidToFirstLogRecord.get(tid.getId()) != null) {
            System.err.print("logXactionBegin: already began this tid\n");
            throw new IOException("double logXactionBegin()");
        }
        preAppend();
        raf.writeInt(BEGIN_RECORD);
        raf.writeLong(tid.getId());
        raf.writeLong(currentOffset);
        tidToFirstLogRecord.put(tid.getId(), currentOffset);
        currentOffset = raf.getFilePointer();
    }

    /**
     * 记录一个检查点日志，并刷新所有页面到磁盘。
     * 锁定缓冲池，强制刷新所有页面，记录当前活跃事务及其首次日志记录位置。
     * @throws IOException
     */
    public void logCheckpoint() throws IOException {
        synchronized (Database.getBufferPool()) {
            synchronized (this) {
                preAppend();
                long startCpOffset = raf.getFilePointer();
                force();
                Database.getBufferPool().flushAllPages();
                raf.writeInt(CHECKPOINT_RECORD);
                raf.writeLong(-1); //no tid , but leave space for convenience

                Set<Long> keys = tidToFirstLogRecord.keySet();
                raf.writeInt(keys.size());
                for (Long key : keys) {
                    raf.writeLong(key);
                    raf.writeLong(tidToFirstLogRecord.get(key));
                }

                long endCpOffset = raf.getFilePointer();
                raf.seek(0);
                raf.writeLong(startCpOffset);
                raf.seek(endCpOffset);
                raf.writeLong(currentOffset);
            }
        }
        logTruncate();
    }

    /**
     * 根据最近的检查点截断日志文件以减少其占用空间。
     * 找到最新的检查点位置，重写日志文件只保留必要的部分，删除旧日志文件并将新文件重命名为原始文件名。
     * @throws IOException
     */
    public synchronized void logTruncate() throws IOException {
        preAppend();
        raf.seek(0);
        long cpLoc = raf.readLong();

        long minLogRecord = cpLoc;

        if (cpLoc != -1L) {
            raf.seek(cpLoc);
            int cpType = raf.readInt();
            @SuppressWarnings("unused")
            long cpTid = raf.readLong();

            if (cpType != CHECKPOINT_RECORD) {
                throw new RuntimeException("Checkpoint pointer does not point to checkpoint record");
            }

            int numOutstanding = raf.readInt();

            for (int i = 0; i < numOutstanding; i++) {
                @SuppressWarnings("unused")
                long tid = raf.readLong();
                long firstLogRecord = raf.readLong();
                if (firstLogRecord < minLogRecord) {
                    minLogRecord = firstLogRecord;
                }
            }
        }

        // 我们可以截断minLogRecord之前的所有内容
        File newFile = new File("logtmp" + System.currentTimeMillis());
        RandomAccessFile logNew = new RandomAccessFile(newFile, "rw");
        logNew.seek(0);
        logNew.writeLong((cpLoc - minLogRecord) + LONG_SIZE);

        raf.seek(minLogRecord);

        // 因为截断后偏移量不同，所以需要重写日志记录
        while (true) {
            try {
                int type = raf.readInt();
                long record_tid = raf.readLong();
                long newStart = logNew.getFilePointer();

                Debug.log("NEW START = " + newStart);

                logNew.writeInt(type);
                logNew.writeLong(record_tid);

                switch (type) {
                    case UPDATE_RECORD:
                        Page before = readPageData(raf);
                        Page after = readPageData(raf);

                        writePageData(logNew, before);
                        writePageData(logNew, after);
                        break;
                    case CHECKPOINT_RECORD:
                        int numXactions = raf.readInt();
                        logNew.writeInt(numXactions);
                        while (numXactions-- > 0) {
                            long xid = raf.readLong();
                            long xoffset = raf.readLong();
                            logNew.writeLong(xid);
                            logNew.writeLong((xoffset - minLogRecord) + LONG_SIZE);
                        }
                        break;
                    case BEGIN_RECORD:
                        tidToFirstLogRecord.put(record_tid, newStart);
                        break;
                }

                // 所有事务在结束时都会有一个指针
                logNew.writeLong(newStart);
                raf.readLong();

            } catch (EOFException e) {
                break;
            }
        }

        Debug.log("TRUNCATING LOG; WAS " + raf.length() + " BYTES ; NEW START : " + minLogRecord + " NEW LENGTH: "
                  + (raf.length() - minLogRecord));

        raf.close();
        logFile.delete();
        newFile.renameTo(logFile);
        raf = new RandomAccessFile(logFile, "rw");
        raf.seek(raf.length());
        newFile.delete();

        currentOffset = raf.getFilePointer();
    }

    /**
     * 回滚指定事务，将该事务更新的所有页面恢复到其更新前的状态。
     * 从日志文件中找到事务的第一个记录位置开始读取，如果遇到的是UPDATE记录且属于指定事务，则使用其“之前”的页面数据覆盖当前页面数据。
     * @param tid
     * @throws NoSuchElementException
     * @throws IOException
     */
    public void rollback(TransactionId tid) throws NoSuchElementException, IOException {
        synchronized (Database.getBufferPool()) {
            synchronized (this) {
                preAppend();
                final Long firstRecordPos = this.tidToFirstLogRecord.get(tid.getId());
                this.raf.seek(firstRecordPos);
                final HashSet<PageId> set = new HashSet<>();
                while (true) {
                    try {
                        final int type = raf.readInt();
                        final long transactionId = raf.readLong();
                        switch (type) {
                            case UPDATE_RECORD: {
                                final Page beginPage = readPageData(this.raf);
                                readPageData(this.raf); // Discard after page data
                                final PageId pageId = beginPage.getId();
                                if (transactionId == tid.getId() && !set.contains(pageId)) {
                                    set.add(pageId);
                                    Database.getBufferPool().discardPage(beginPage.getId());
                                    Database.getCatalog().getDatabaseFile(pageId.getTableId()).writePage(beginPage);
                                }
                                break;
                            }
                            case CHECKPOINT_RECORD:
                                skipCheckPointRecord();
                                break;
                        }
                        raf.readLong(); // Read record start offset
                    } catch (final EOFException e) {
                        break;
                    }
                }
            }
        }
    }

    /**
     * 跳过检查点记录中的信息，因为不需要处理它们。
     * 根据活跃事务数量计算需要跳过的字节数，并执行跳过操作。
     * @throws IOException
     */
    private void skipCheckPointRecord() throws IOException {
        final int txnCnt = this.raf.readInt();
        final int skip = txnCnt * 2 * 8; // Each transaction has a long ID and a long offset
        this.raf.skipBytes(skip);
    }

    /**
     * 通过确保已提交事务的更新被安装并且未提交事务的更新不被安装来恢复数据库系统
     * 遍历日志文件，分别收集每个事务的“之前”和“之后”的页面数据。
     * 对于未提交的事务，使用“之前”的页面数据进行恢复；对于已提交的事务，使用“之后”的页面数据更新数据库。
     * @throws IOException
     */
    public void recover() throws IOException {
        synchronized (Database.getBufferPool()) {
            synchronized (this) {
                recoveryUndecided = false;
                this.raf.seek(0);
                final long cp = raf.readLong();
                if (cp > 0) {
                    this.raf.seek(cp);
                }
                final HashSet<Long> commitIds = new HashSet<>();
                final HashMap<Long, List<Page>> beforePages = new HashMap<>();
                final HashMap<Long, List<Page>> afterPages = new HashMap<>();
                while (true) {
                    try {
                        final int type = this.raf.readInt();
                        final long tid = this.raf.readLong();
                        switch (type) {
                            case UPDATE_RECORD: {
                                final Page beforePage = readPageData(raf);
                                final Page afterPage = readPageData(raf);

                                beforePages.computeIfAbsent(tid, k -> new ArrayList<>()).add(beforePage);
                                afterPages.computeIfAbsent(tid, k -> new ArrayList<>()).add(afterPage);
                                break;
                            }
                            case COMMIT_RECORD:
                                commitIds.add(tid);
                                break;
                            case CHECKPOINT_RECORD:
                                skipCheckPointRecord();
                                break;
                        }
                    } catch (EOFException e) {
                        break;
                    }
                }
                // Roll back uncommitted transactions
                beforePages.forEach((tid, pages) -> {
                    if (!commitIds.contains(tid)) {
                        for (final Page page : pages) {
                            try {
                                Database.getCatalog().getDatabaseFile(page.getId().getTableId()).writePage(page);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                });
                // Write committed transactions' pages
                for (final Long commitId : commitIds) {
                    if (afterPages.containsKey(commitId)) {
                        final List<Page> pages = afterPages.get(commitId);
                        for (final Page page : pages) {
                            Database.getCatalog().getDatabaseFile(page.getId().getTableId()).writePage(page);
                        }
                    }
                }
            }
        }
    }

    /**
     * 关闭日志系统，写出必要的状态以便快速启动（无需大量恢复）。
     * 写入一个检查点记录然后关闭随机访问文件对象。
     */
    public synchronized void shutdown() {
        try {
            logCheckpoint(); // Writing a checkpoint record is a simple way to shut down
            raf.close();
        } catch (IOException e) {
            System.out.println("ERROR SHUTTING DOWN -- IGNORING.");
            e.printStackTrace();
        }
    }

    /**
     * 打印出日志的人类可读表示。
     * 从头开始读取日志文件并打印每条记录的类型、事务ID和其他相关信息。
     * @throws IOException
     */
    public void print() throws IOException {
        long curOffset = raf.getFilePointer(); // 保存当前文件指针的位置

        raf.seek(0); // 将文件指针移动到文件开头

        System.out.println("0: checkpoint record at offset " + raf.readLong()); // 打印检查点记录的位置

        while (true) { // 开始遍历日志文件
            try {
                int cpType = raf.readInt(); // 读取记录类型
                long cpTid = raf.readLong(); // 读取事务ID

                System.out.println((raf.getFilePointer() - (INT_SIZE + LONG_SIZE)) + ": RECORD TYPE " + cpType);
                System.out.println((raf.getFilePointer() - LONG_SIZE) + ": TID " + cpTid);

                switch (cpType) { // 根据记录类型进行不同的处理
                    case BEGIN_RECORD:
                        System.out.println(" (BEGIN)");
                        System.out.println(raf.getFilePointer() + ": RECORD START OFFSET: " + raf.readLong());
                        break;
                    case ABORT_RECORD:
                        System.out.println(" (ABORT)");
                        System.out.println(raf.getFilePointer() + ": RECORD START OFFSET: " + raf.readLong());
                        break;
                    case COMMIT_RECORD:
                        System.out.println(" (COMMIT)");
                        System.out.println(raf.getFilePointer() + ": RECORD START OFFSET: " + raf.readLong());
                        break;

                    case CHECKPOINT_RECORD:
                        System.out.println(" (CHECKPOINT)");
                        int numTransactions = raf.readInt();
                        System.out.println((raf.getFilePointer() - INT_SIZE) + ": NUMBER OF OUTSTANDING RECORDS: "
                                           + numTransactions);

                        while (numTransactions-- > 0) { // 遍历未完成的事务
                            long tid = raf.readLong();
                            long firstRecord = raf.readLong();
                            System.out.println((raf.getFilePointer() - (LONG_SIZE + LONG_SIZE)) + ": TID: " + tid);
                            System.out.println((raf.getFilePointer() - LONG_SIZE) + ": FIRST LOG RECORD: "
                                               + firstRecord);
                        }
                        System.out.println(raf.getFilePointer() + ": RECORD START OFFSET: " + raf.readLong());

                        break;
                    case UPDATE_RECORD:
                        System.out.println(" (UPDATE)");

                        long start = raf.getFilePointer();
                        Page before = readPageData(raf);

                        long middle = raf.getFilePointer();
                        Page after = readPageData(raf);

                        System.out.println(start + ": before image table id " + before.getId().getTableId());
                        System.out.println((start + INT_SIZE) + ": before image page number "
                                           + before.getId().getPageNumber());
                        System.out.println((start + INT_SIZE) + " TO " + (middle - INT_SIZE) + ": page data");

                        System.out.println(middle + ": after image table id " + after.getId().getTableId());
                        System.out.println((middle + INT_SIZE) + ": after image page number "
                                           + after.getId().getPageNumber());
                        System.out.println((middle + INT_SIZE) + " TO " + (raf.getFilePointer()) + ": page data");

                        System.out.println(raf.getFilePointer() + ": RECORD START OFFSET: " + raf.readLong());

                        break;
                }

            } catch (EOFException e) { // 如果到达文件末尾，则退出循环
                break;
            }
        }

        raf.seek(curOffset); // 恢复文件指针到原来的位置
    }

    /**
     * 该方法使用RandomAccessFile对象的通道(Channel)的force(boolean)方法来确保所有对文件的更改都被物理写入到磁盘
     * 参数true表示同步文件元数据
     * @throws IOException
     */
    public synchronized void force() throws IOException {
        raf.getChannel().force(true);
    }
}
